package main

import (
	"encoding/json"
	"flag"
	"fmt"
	"go_19/core"
	"go_19/global"
	"go_19/service/pulsar/pulsar_topic_mover"
	"time"
)

// pulsar topic data move
func main() {
	core.EmptyCli()
	topicMover := getParams()
	jsonStr, _ := json.Marshal(topicMover)
	fmt.Println(string(jsonStr))
	pulsar_topic_mover.TopicMover(topicMover)

}
// 获取cli参数
func getParams() pulsar_topic_mover.TopicMoverStruct {
	fmt.Println("Pulsar Subscribe A Topic =>B Topic")
	var param pulsar_topic_mover.TopicMoverStruct
	fromPulsarBroker := flag.String("from_pulsar_broker", "", "required A Topic Pulsar Broker")
	fromPulsarTopic := flag.String("from_pulsar_topic", "", "required A Topic")
	fromPulsarTopicSubscriptionInitialPosition := flag.Int("from_pulsar_topic_subscription_initial_position", 0, "Not required default 0 【0:latest,1:earliest】")
	toPulsarBroker := flag.String("to_pulsar_broker", "", "required B Topic Pulsar Broker")
	toPulsarTopic := flag.String("to_pulsar_topic", "", "required B Topic")
	toPulsarTopicDelay := flag.Int("to_pulsar_topic_delay", 0, "Not required default 0 【0:no delay,>0:delay to topic by X seconds】")
	flag.Parse()

	if *fromPulsarBroker == "" {
		panic("please input from_pulsar_broker!")
	}
	param.FromTopic.ClientOption.URL = *fromPulsarBroker
	param.FromTopic.ClientOption.ConnectionTimeout = time.Second * time.Duration(5)
	param.FromTopic.ClientOption.OperationTimeout = time.Second * time.Duration(30)
	param.FromTopic.ClientOption.MaxConnectionsPerBroker = 1
	param.FromTopic.ClientOption.LogPath = global.GVA_CONFIG.Log.LogDir
	param.FromTopic.ClientOption.LogName = "fromTopic_" + *fromPulsarTopic
	param.FromTopic.SubscriptionType = 0
	param.FromTopic.SubscriptionName = "pulsar_topic_mover"
	param.FromTopic.ConsumerName = "pulsar_topic_mover"
	param.FromTopic.SubscriptionInitialPosition = *fromPulsarTopicSubscriptionInitialPosition

	if *fromPulsarTopic == "" {
		panic("please input from_pulsar_topic!")
	}
	param.FromTopic.Topic = *fromPulsarTopic

	if *toPulsarBroker == "" {
		panic("please input to_pulsar_broker!")
	}
	param.ToTopic.ClientOption.URL = *toPulsarBroker
	param.ToTopic.ClientOption.LogPath = global.GVA_CONFIG.Log.LogDir
	param.ToTopic.ClientOption.ConnectionTimeout = time.Second * time.Duration(5)
	param.ToTopic.ClientOption.OperationTimeout = time.Second * time.Duration(30)
	param.ToTopic.ClientOption.MaxConnectionsPerBroker = 1

	if *toPulsarTopic == "" {
		panic("please input to_pulsar_topic!!")
	}
	param.ToTopic.Topic = *toPulsarTopic
	param.ToTopic.ClientOption.LogName = "toTopic_" + *toPulsarTopic
	param.ToTopic.DeliverAfter = *toPulsarTopicDelay
	param.ToTopic.ProducerName = "pulsar_topic_mover_producer"
	return param
}
